//====================================================================
// Filename:      goblin_sqs_queue.cc
//                
// It belongs to goblin-os.
//                
// Version:       
// Author:        Ins <InsFous@gmail.com>
// Created at:    Fri Dec 10 17:50:54 2010
// Description:   
//                
//====================================================================
#include "goblin_sqs_queue.h"

#include <fcntl.h>

namespace goblin {
  namespace sqs {
    Queue::Queue(const QueueName &queue_name, const std::string &db_dir):
      unread_msg_count_(0), unsync_to_disk_msg_count_(0), current_max_msg_id_(0), capacity_(kQueueDefaultCapacity),
      filter_(-1), unsync_to_disk_msgs_max_msg_id_(0), db_num_(1), last_db_index_(1)
    {
      unread_msgs_ = (struct QueueNode *)malloc(sizeof(struct QueueNode));
      unsync_to_disk_msgs_ = (struct QueueNode *)malloc(sizeof(struct QueueNode));
      INIT_LIST_HEAD(&(unread_msgs_->list));
      INIT_LIST_HEAD(&(unsync_to_disk_msgs_->list));
      queue_name_ = queue_name;
      db_dir_ = db_dir;
      pthread_mutex_init(&mutex_lock_, NULL);
      Init();
    }
    Queue::~Queue() {
      pthread_mutex_destroy(&mutex_lock_);
    }
    
    void Queue::Init() {
      ReadMetaData();
      ReadinUnreadMsgs();
    }
    bool Queue::PushMessage(const std::string &msg_content, const int32_t msg_type) {
      Message *msg = new Message();
      msg->msg_id = AllocMessageID();
      msg->msg_content.append(msg_content);
      msg->msg_insert_time = time(NULL);
      msg->msg_type = msg_type;
      struct QueueNode *node = (struct QueueNode *)malloc(sizeof(struct QueueNode));
      node->msg = msg;
      ++unread_msg_count_;
      pthread_mutex_lock(&mutex_lock_);
      if (unread_msg_count_ > capacity_) {
	PopMessage();
      }
      printf("PushMessage msg id:%lld\n", msg->msg_id);
      list_add_tail(&(node->list), &(unread_msgs_->list));
      pthread_mutex_unlock(&mutex_lock_);
      return true;
    }
    bool Queue::GetMessage(Message &msg) {
      pthread_mutex_lock(&mutex_lock_);
      while (unread_msg_count_) {
	struct QueueNode *node = list_entry(unread_msgs_->list.next, struct QueueNode, list);
	Message *m = node->msg;
	if (filter_ != -1 && filter_ != m->msg_type) {
	  list_del(&(node->list));
	  list_add_tail(&(node->list), &(unsync_to_disk_msgs_->list));
	  --unread_msg_count_;
	  ++unsync_to_disk_msg_count_;
	  ++unsync_to_disk_msgs_max_msg_id_;
	  continue;
	} else {
	  msg.msg_id = m->msg_id;
	  msg.msg_insert_time = m->msg_insert_time;
	  msg.msg_content = m->msg_content;
	  msg.msg_type = m->msg_type;
	  list_del(&(node->list));
	  list_add_tail(&(node->list), &(unsync_to_disk_msgs_->list));
	  --unread_msg_count_;
	  ++unsync_to_disk_msg_count_;
	  ++unsync_to_disk_msgs_max_msg_id_;
	  pthread_mutex_unlock(&mutex_lock_);
	  return true;
	}
      }
      msg.msg_id = -1;
      pthread_mutex_unlock(&mutex_lock_);
      return false;
    }
    bool Queue::PopMessage() {
      if (unread_msg_count_) {
	struct QueueNode *node = list_entry(&(unread_msgs_->list.next), struct QueueNode, list);
	list_del(&(node->list));
	list_add_tail(&(node->list), &(unsync_to_disk_msgs_->list));
	--unread_msg_count_;
	++unsync_to_disk_msg_count_;
	++unsync_to_disk_msgs_max_msg_id_;
	return true;
      }
      return false;
    }
    bool Queue::SetCapacity(const int64_t capacity) {
      if (capacity <= 0) return false;
      pthread_mutex_lock(&mutex_lock_);
      if (capacity >= capacity_ || capacity >= unread_msg_count_) {
	capacity_ = capacity;
	pthread_mutex_unlock(&mutex_lock_);
	return true;
      } else {
	while (unread_msg_count_ != capacity) {
	  PopMessage();
	}
	capacity_ = capacity;
	pthread_mutex_unlock(&mutex_lock_);
	return true;
      }
    }

    void Queue::WriteMetaData() {
      char path[1024] = {0};
      snprintf(path, sizeof(path), "%s/%s-metadata", db_dir_.c_str(), queue_name_.c_str());
      struct DBMetaData metadata;
      pthread_mutex_lock(&mutex_lock_);
      metadata.db_num = db_num_;
      metadata.last_db_index = last_db_index_;
      metadata.max_msg_id = unsync_to_disk_msgs_max_msg_id_;
      pthread_mutex_unlock(&mutex_lock_);
      FILE *fp = fopen(path, "wb");
      if (fp != NULL) {
	fwrite((char *)&metadata, 1, sizeof(struct DBMetaData), fp);
	fflush(fp);
	fclose(fp);
      } else {
	printf("Path Not Exists: {%s}\n", path);
      }
    }

    void Queue::ReadMetaData() {
      char path[1024] = {0};
      snprintf(path, sizeof(path), "%s/%s-metadata", db_dir_.c_str(), queue_name_.c_str());
      FILE *fp = fopen(path, "rb");
      if (fp == NULL) {
	printf("meta not exits, go on\n");
      } else {
	char *buf = (char *)malloc(sizeof(struct DBMetaData));
	fread(buf, 1, sizeof(struct DBMetaData), fp);
	struct DBMetaData *metadata = (struct DBMetaData *)buf;
	pthread_mutex_lock(&mutex_lock_);
	db_num_ = metadata->db_num;
	last_db_index_ = metadata->last_db_index;
	unsync_to_disk_msgs_max_msg_id_ = metadata->max_msg_id;
	current_max_msg_id_ = unsync_to_disk_msgs_max_msg_id_;
	pthread_mutex_unlock(&mutex_lock_);
	free(buf);
	fclose(fp);
      }
    }

    void Queue::WriteUnSyncMsgs() {
      char path[1024] = {0};
      snprintf(path, sizeof(path), "%s/%s-%lld", db_dir_.c_str(), queue_name_.c_str(), last_db_index_);
      printf("Sync DB File:{%s}\n", path);
      int fd = open(path, O_RDWR | O_CREAT, 0666);
      long pos = lseek(fd, 0L, SEEK_END);
      struct DBHeader header;
      memset(&header, 0, sizeof(struct DBHeader));
      if( pos != 0) {
	lseek(fd, 0L, SEEK_SET);
	read(fd, (char *)&header, sizeof(struct DBHeader));
	lseek(fd, header.begin_write_offset, SEEK_SET);
      } else {
	write(fd, (char *)&header, sizeof(struct DBHeader));
      }
      pthread_mutex_lock(&mutex_lock_);
      while (!list_empty(&(unsync_to_disk_msgs_->list))) {
	struct list_head *head = unsync_to_disk_msgs_->list.next;
	struct QueueNode *node = list_entry(head, struct QueueNode, list);
	printf("MSGID: %lld\n", node->msg->msg_id);
	struct DBMessage *msg = (struct DBMessage *)malloc(sizeof(struct DBMessage) + node->msg->msg_content.length());
	msg->msg_id = node->msg->msg_id;
	msg->msg_insert_time = node->msg->msg_insert_time;
	msg->msg_type = node->msg->msg_type;
	msg->msg_content_len = node->msg->msg_content.length();
	memcpy(msg->msg_content, node->msg->msg_content.c_str(), msg->msg_content_len);
	write(fd, (char *)msg, sizeof(struct DBMessage) + msg->msg_content_len);
	free(msg);
	delete node->msg;
	list_del(head);
	free(node);
      }
      unsync_to_disk_msg_count_ = 0;
      pos = lseek(fd, 0L, SEEK_END);
      header.begin_write_offset = pos;
      header.last_msg_id = current_max_msg_id_;
      pthread_mutex_unlock(&mutex_lock_);
      lseek(fd, 0L, SEEK_SET);
      write(fd, (char *)&header, sizeof(struct DBHeader));
      close(fd);
    }

    void Queue::SyncToDisk() {
      WriteUnSyncMsgs();
      WriteMetaData();
    }
    
    void Queue::WriteUnreadMsgs() {
      char path[1024] = {0};
      snprintf(path, sizeof(path), "%s/%s-unread", db_dir_.c_str(), queue_name_.c_str());
      FILE *fp = fopen(path, "wb");
      if (fp == NULL) return;
      pthread_mutex_lock(&mutex_lock_);
      while (!list_empty(&(unread_msgs_->list))) {
	struct list_head *head = unread_msgs_->list.next;
	struct QueueNode *node = list_entry(head, struct QueueNode, list);
	printf("Write Unread MSG ID: %lld\n", node->msg->msg_id);
	struct DBMessage *msg = (struct DBMessage *)malloc(sizeof(struct DBMessage) + node->msg->msg_content.length());
	msg->msg_id = node->msg->msg_id;
	msg->msg_insert_time = node->msg->msg_insert_time;
	msg->msg_type = node->msg->msg_type;
	msg->msg_content_len = node->msg->msg_content.length();
	memcpy(msg->msg_content, node->msg->msg_content.c_str(), msg->msg_content_len);
	int n = fwrite((char *)msg, 1, sizeof(struct DBMessage) + msg->msg_content_len, fp);
	free(msg);
	delete node->msg;
	list_del(head);
	free(node);
      }
      pthread_mutex_unlock(&mutex_lock_);
      fclose(fp);
    }

    void Queue::ReadinUnreadMsgs() {
      char path[1024] = {0};
      snprintf(path, sizeof(path), "%s/%s-unread", db_dir_.c_str(), queue_name_.c_str());
      FILE *fp = fopen(path, "rb");
      if (fp != NULL) {
	std::string content;
	int n = -1;
	char buf[1024] = {0};
	while ((n = fread(buf, 1, sizeof(buf), fp)) > 0) {
	  content.append(buf, n);
	}
	const char *ptr= content.c_str();
	long left = content.length();
	while (left) {
	  struct DBMessage *message = (struct DBMessage *)ptr;
	  Message *msg = new Message();
	  msg->msg_id = message->msg_id;
	  printf("Import Unread Msg %lld\n", msg->msg_id);
	  msg->msg_content.append(message->msg_content, message->msg_content_len);
	  msg->msg_insert_time = message->msg_insert_time;
	  msg->msg_type = message->msg_type;
	  struct QueueNode *node = (struct QueueNode *)malloc(sizeof(struct QueueNode));
	  node->msg = msg;
	  ++unread_msg_count_;
	  if (unread_msg_count_ > capacity_) {
	    PopMessage();
	  }
	  list_add_tail(&(node->list), &(unread_msgs_->list));
	  current_max_msg_id_ = msg->msg_id;
	  ptr += sizeof(struct DBMessage) + message->msg_content_len;
	  left -= sizeof(struct DBMessage) + message->msg_content_len;
	}
	fclose(fp);
      }
    }

    void Queue::Save() {
      SyncToDisk();
      WriteUnreadMsgs();
    }
  }
}
